/* =========================================
 *           消息转发子服务实现 
 * =========================================*/
#pragma once
#include "channel.hpp"
#include "rabbitmq.hpp"
#include "odb_factory.hpp"
#include "chat_session_member_dao.hpp"
#include "etcd.hpp"
#include "logger.hpp"
#include "base.pb.h"
#include "message_transmit.pb.h"
#include "user.pb.h"
#include "utils.hpp"
#include <brpc/server.h>
#include <ctime>
#include <memory>
#include <string>

namespace im_server {

class MsgTransmitServiceImpl;
class MsgTransmitServer;
class MsgTransmitServerBuilder;
using MsgTransmitServiceImplPtr = std::shared_ptr<MsgTransmitServiceImpl>;
using MsgTransmitServerPtr = std::shared_ptr<MsgTransmitServer>;
using MsgTransmitServerBuilderPtr = std::shared_ptr<MsgTransmitServerBuilder>;

class MsgTransmiteServiceImpl : public im_proto::MsgTransmitService
{
public:
    MsgTransmiteServiceImpl(const shared::dal::OdbPtr& mysqlClient,
        const shared::infra::RpcServiceManagerPtr& rpcServiceMgr,
        const shared::infra::MQClientPtr& mqClient,
        const std::string& exchange, const std::string& routingKey,
        const std::string& userServiceName) 
        : chatSessionMemberDao_(std::make_shared<shared::dal::ChatSessionMemberDao>(mysqlClient))
        , mqClient_(mqClient), exchange_(exchange), routingKey_(routingKey)
        , rpcServiceMgr_(rpcServiceMgr)
        , userServiceName_(userServiceName) {}
    ~MsgTransmiteServiceImpl() {}

    void GetTransmitTarget(::google::protobuf::RpcController* controller,
                        const ::im_proto::NewMessageReq* request,
                        ::im_proto::GetTransmitTargetRsp* response,
                        ::google::protobuf::Closure* done)
    {
        brpc::ClosureGuard rpc_guard(done);
        const std::string& rid = request->request_id();
        response->set_request_id(rid);

        auto errResp = [&](const std::string errmsg) {
            response->set_success(false);
            response->set_errmsg(errmsg);
        };

        // 1. 获取会话信息
        const std::string& userId = request->user_id();
        const std::string& chatSessionId = request->chat_session_id();
        // 2. rpc通信获取该消息的所属用户信息
        auto channel = rpcServiceMgr_->choose(userServiceName_);
        if(!channel) {
            LOG_ERROR("{}-{}-{} 用户子服务节点不存在!", rid, userId, userServiceName_);
            return errResp("用户子服务节点不存在");
        }  
        // 2-1 构造请求体
        im_proto::GetUserInfoReq req;
        req.set_request_id(rid);
        req.set_user_id(userId);
        // 2-2 通过信道发起服务请求
        brpc::Controller ctl;
        im_proto::GetUserInfoRsp resp;
        im_proto::UserService_Stub stub(channel.get());
        stub.GetUserInfo(&ctl, &req, &resp, nullptr);
        if(ctl.Failed() || !resp.success()) {
            LOG_ERROR("{}-{} 获取用户信息失败!", rid, ctl.ErrorText());
            return errResp("获取用户信息失败");
        } 
        // 3. 该消息的所有目标用户不一定都在线, 因此需要发布到消息队列, 由消费者决定何时消费(用户上线时消费)
        // 3-1 构建转发的消息完整结构
        im_proto::MessageInfo messageInfo;
        messageInfo.set_message_id(shared::Uuid::createUuid());
        messageInfo.set_chat_session_id(chatSessionId);
        messageInfo.set_timestamp(time(nullptr));
        messageInfo.mutable_sender()->CopyFrom(resp.user_info());
        messageInfo.mutable_message()->CopyFrom(request->message());
        // 3-2 查表, 获取该消息转发的所有目标用户
        auto targetUserList = chatSessionMemberDao_->selectMembers(chatSessionId);
        // 3-3 发布消息
        if(!mqClient_->publish(exchange_, messageInfo.SerializeAsString(), routingKey_)) {
            LOG_ERROR("{}-持久化消息发布失败: {}!", request->request_id(), ctl.ErrorText());
            return errResp("持久化消息发布失败!");
        }
        // 4. 构建消息转发的响应, 由网关进行转发消息
        response->set_success(true);
        response->mutable_message()->CopyFrom(messageInfo);
        for(const auto& userId : targetUserList) {
            response->add_target_id_list(userId);
        }
    }

private:
    shared::dal::ChatSessionMemberDaoPtr chatSessionMemberDao_;   // 持久化数据的操作对象

    shared::infra::MQClientPtr mqClient_;   // 消息队列客户端句柄
    std::string exchange_;     
    std::string routingKey_;

    shared::infra::RpcServiceManagerPtr rpcServiceMgr_;           // rpc的服务管理对象

    std::string userServiceName_;   // 用户子服务名称, 通过rpc获取用户信息
};

class MsgTransmitServer
{
public:
    MsgTransmitServer(const shared::dal::OdbPtr& mysqlClient,
                const std::shared_ptr<brpc::Server>& rpcServer,
                const shared::infra::DiscoveryPtr& rpcSvrDisc, 
                const shared::infra::RegistrantPtr& rpcSvrRegist)
        : mysqlClient_(mysqlClient), rpcServer_(rpcServer)
        , rpcSvrDisc_(rpcSvrDisc), rpcSvrRegist_(rpcSvrRegist) {}
    ~MsgTransmitServer() {}
    void start() { rpcServer_->RunUntilAskedToQuit(); }
private:
    shared::dal::OdbPtr mysqlClient_;       // 聊天会话成员的数据持久化句柄

    // 这三个对象都不是直接面向业务功能的, 而是为业务功能提供服务支撑
    // 因此服务没有结束, 这三个对象就不能释放, 于是放在MsgTransmitServer中
    std::shared_ptr<brpc::Server> rpcServer_;        // rpc服务器
    shared::infra::DiscoveryPtr rpcSvrDisc_;         // rpc服务发现对象
    shared::infra::RegistrantPtr rpcSvrRegist_;      // rpc服务注册对象
};

class MsgTransmitServerBuilder
{
public:
    // 构造消息队列客户端对象
    void makeRabbitMQObject(const std::string& user, const std::string& passwd, const std::string& host,
        const std::string& exchange, const std::string& queue, const std::string& bindingKey)
    {
        mqClient_ = std::make_shared<shared::infra::MQClient>(user, passwd, host);
        mqClient_->declareComponents(exchange, queue);
        exchange_ = exchange;
        routingKey_ = bindingKey;
    }
    // 构造odb-mysql客户端对象
    void makeMysqlObject(const std::string& MsgTransmit,  const std::string& passwd, 
        const std::string& host, const std::string& db,
        const std::string& cset, uint16_t port, uint16_t connectPoolCount)
    {
        mysqlClient_ = shared::dal::ODBFactory::create(
            MsgTransmit, passwd, host, db, cset, port, connectPoolCount
        );
    }
    // 构造Rpc服务注册客户端对象
    void makeRpcSvrRegistrant(const std::string& host, const std::string& serviceName, const std::string& accessHost)
    {
        rpcSvrRegist_ = std::make_shared<shared::infra::Registrant>(host);
        rpcSvrRegist_->registry(serviceName, accessHost);
    }
    // 构造Rpc服务发现客户端对象
    void makeRpcSvrDiscoverer(const std::string& host, const std::string& baseServiceName, const std::string& userServiceName)
    {
        // 用户管理子服务中需要关注文件子服务, 用于上传下载用户头像数据
        userServiceName_ = userServiceName;     
        rpcServiceMgr_ = std::make_shared<shared::infra::RpcServiceManager>();
        rpcServiceMgr_->declared(userServiceName);

        auto putCb = std::bind(
            &shared::infra::RpcServiceManager::onServiceOnline, rpcServiceMgr_.get(),
            std::placeholders::_1, std::placeholders::_2
        );
        auto delCb = std::bind(
            &shared::infra::RpcServiceManager::onServiceOffline, rpcServiceMgr_.get(),
            std::placeholders::_1, std::placeholders::_2
        );
        rpcSvrDisc_ = std::make_shared<shared::infra::Discovery>(host, baseServiceName, putCb, delCb);
    }
    // 构造RPC服务器对象
    void makeRpcServer(uint16_t port, int32_t timeout, uint8_t numThreads)
    {
        if (!mysqlClient_) {
            LOG_ERROR("还未初始化mysql持久化模块!");
            abort();
        }
        if(!mqClient_) {
            LOG_ERROR("还未初始化rabbitMQ消息队列模块!");
            abort();
        }
        if (!rpcServiceMgr_) {
            LOG_ERROR("还未初始化rpc服务管理模块!");
            abort();
        }
        rpcServer_ = std::make_shared<brpc::Server>();
        
        // 向rpc服务器添加服务实现类
        MsgTransmiteServiceImpl* msgTransmiteServiceImpl = new MsgTransmiteServiceImpl(
            mysqlClient_, rpcServiceMgr_, mqClient_, exchange_, routingKey_, userServiceName_
        );
        if(rpcServer_->AddService(msgTransmiteServiceImpl, brpc::ServiceOwnership::SERVER_OWNS_SERVICE) == -1) {
            LOG_ERROR("添加Rpc服务失败!");
            abort();
        }
        // 启动服务器
        brpc::ServerOptions options;    
        options.idle_timeout_sec = timeout;  
        options.num_threads = numThreads;        
        if(rpcServer_->Start(port, &options) == -1) {
            LOG_ERROR("Rpc服务启动失败!");
            abort();
        }
    }
    MsgTransmitServerPtr build()
    {
        if(!rpcSvrDisc_) {
            LOG_ERROR("还未初始化服务发现模块!");
            abort();
        }
        if (!rpcSvrRegist_) {
            LOG_ERROR("还未初始化服务注册模块!");
            abort();
        }
        if (!rpcServer_) {
            LOG_ERROR("还未初始化RPC服务器模块!");
            abort();
        }
        return std::make_shared<MsgTransmitServer>(
            mysqlClient_, rpcServer_, rpcSvrDisc_, rpcSvrRegist_
        );
    }
private:
    shared::dal::OdbPtr mysqlClient_;       // 用户数据的持久化句柄

    std::shared_ptr<brpc::Server> rpcServer_;               // rpc服务器
    shared::infra::RpcServiceManagerPtr rpcServiceMgr_;     // rpc的服务管理对象
    shared::infra::RegistrantPtr rpcSvrRegist_;             // rpc服务注册对象
    shared::infra::DiscoveryPtr rpcSvrDisc_;                // rpc服务发现对象
    
    shared::infra::MQClientPtr mqClient_;   // 消息队列客户端句柄
    std::string exchange_;                   // 发布消息所需参数
    std::string routingKey_;

    std::string userServiceName_;   // 文件子服务名称, 通过rpc获取用户头像数据
};



}